package com.study.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming05_State {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc =new StreamingContext(sparkConf,Seconds(3))
    ssc.checkpoint("cp")

    //无状态的数据操作，只对当前的采集周期内的数据进行处理
    //在某些场合下，需要保留数据统计结果（状态），实现数据汇总
    //使用有状态操作时，需要设置检查点路径
    val datas = ssc.socketTextStream("localhost", 9999)
    val wordToOne: DStream[(String, Int)] = datas.map((_, 1))
//    val wordToCount = wordToOne.reduceByKey(_ + _)

    //updateStateByKey:根据key对数据状态进行更新
    //传递的参数：第一个表示相同key的value的数据  第二个表示缓存去相同key的value数据
    val state = wordToOne.updateStateByKey(
      (seq: Seq[Int], buff: Option[Int]) => {
        val newCount = buff.getOrElse(0) + seq.sum
        Option(newCount)
      }
    )
    state.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
